package com.jd.cmdb.tsc.zk.listener;

import com.jd.cmdb.tsc.TaskStateEnum;
import com.jd.cmdb.tsc.entity.Task;
import com.jd.cmdb.tsc.entity.TaskLog;
import com.jd.cmdb.tsc.zk.exception.ZKExceptionHandler;
import com.jd.cmdb.tsc.service.TaskLogService;
import com.jd.cmdb.tsc.service.TaskService;
import com.jd.cmdb.tsc.util.TaskUtil;
import com.jd.cmdb.tsc.zk.service.ZKService;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

/**
 * Created by tianzhengguo on 2016/1/11.
 */
public class WorkerNodeListener implements PathChildrenCacheListener {
    public static final Logger logger = LoggerFactory.getLogger(WorkerNodeListener.class);

    private TaskService taskService;
    private TaskLogService taskLogService;
    private String tscNodePath;
    private ZKService zkService;
    private LeaderSelector leaderSelector;
    private CountDownLatch leaderLatch;

    public WorkerNodeListener(TaskService taskService, TaskLogService taskLogService,
                              ZKService zkService, CountDownLatch leaderLatch) {
        this.taskService = taskService;
        this.taskLogService = taskLogService;
        this.zkService = zkService;
        this.leaderLatch = leaderLatch;
        this.tscNodePath = zkService.getTSCNode();
        this.leaderSelector = zkService.getLeaderSelector();
    }

    /**
     * Called when a change has occurred
     *
     * @param client the factory
     * @param event  describes the change
     * @throws Exception errors
     */
    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
        try {
            String data = new String(event.getData().getData());
            logger.info("被触发的事件类型:{}",event.getType());

             //如果是注册worker节点，leader监控节点不需要做任何操作。
            if(event.getType().equals(Type.CHILD_ADDED)){
                logger.info("worker向zk中注册节点:{}",event.getData().getPath());
                logger.info("注册节点时,worker节点的数据:{}",new String(event.getData().getData()));
                return ;
            }

            //如果当前的客户端，不是leader
            //则阻塞当前的客户端，直到leader处理完节点状态的操作后，
            //由leader release当前客户端的阻塞
            //在leader处理节点状态过程中，如果leader宕机了，
            //会由curator自己选举出新的leader，选举出leader后，会release掉，对应的客户端，
            //该客户端就成为新的leader，重新执行原leader没有处理完的数据
            if(!zkService.isLeader())
                leaderLatch.await();

            //leader处理完数据后，当前的客户端如果不是leader，则直接返回。
            if(!zkService.isLeader())
                return;

            logger.info("触发事件的worker节点:{}", event.getData().getPath());

            logger.info("worker中的数据:{}",data);

            Map<String,Object> jsonMap = TaskUtil.parseTaskJson(data);
            TaskStateEnum taskState = judgeState(Short.valueOf(jsonMap.get(TaskUtil.TASK_STATUS).toString()));
            short shortState = taskState.getState();

            Task task = (Task) jsonMap.get(TaskUtil.TASK_KEY);

            task.setState(shortState);
            task.setStateDesc(taskState.stateDesc());

            TaskLog taskLog = (TaskLog) jsonMap.get(TaskUtil.LOG_KEY);

            short[] states = null;
            switch (event.getType()) {
                //节点数据更新
                case CHILD_UPDATED: {
                    logger.info("事件类型:{}", event.getType());
                    logger.info("更改任务状态为:{}", taskState.stateDesc());
                    //如果节点中的消息状态是STATE_RUNNING，则表示worker刚处理了
                    //非STATE_RUNNING的数据，task的状态变更为STATE_RUNNING
                    if(shortState == TaskService.STATE_RUNNING){
                        states =new short[] {TaskService.STATE_READY,TaskService.STATE_SUCCESS,
                                TaskService.STATE_WORKER_INNER_FAIL,TaskService.STATE_WORKER_OUTER_FAIL};
                    }else
                    //如果节点的状态是非STATE_RUNNING的，则表示之前的节点状态为STATE_RUNNING
                    //task状态变更为STATE_RUNNING
                    if(shortState == TaskService.STATE_READY
                            ||shortState == TaskService.STATE_SUCCESS
                            ||shortState == TaskService.STATE_WORKER_INNER_FAIL
                            ||shortState == TaskService.STATE_WORKER_OUTER_FAIL){
                        states = new short[]{TaskService.STATE_RUNNING};
                    }
                    break;
                }
                //节点被删除
                //节点被删除时，worker节点会缓存删除之前的数据，
                //如果之前的数据状态为非STATE_RUNNING，节点删除之前的任务表示该任务已正常处理，不需要重发任务。
                //所以不对非STATE_RUNNING的数据进行处理，
                //如果之前的数据是STATE_RUNNING，则表示任务还没有处理完成，
                // 更改任务状态为失败，下次轮询时，重发该任务。
                case CHILD_REMOVED: {
                    logger.info("事件类型:{}", event.getType());
                    logger.info("当前leader;{}", this.leaderSelector.getLeader().getId());
                    task.setState(TaskService.STATE_WORKER_INNER_FAIL);
                    String stateDesc = TaskStateEnum.values()[TaskService.STATE_WORKER_INNER_FAIL].stateDesc();
                    task.setStateDesc(stateDesc);
                    states = new short[] {TaskService.STATE_RUNNING};
                    break;
                }
                default: {
                    break;
                }
            }
            logger.info("待处理数据:\n{}", data);

            resolveTaskJson(task, taskLog, taskState, states);
            logger.info("更改TSCNODE节点的数据,取消非leader节点的阻塞.");
            client.setData().forPath(tscNodePath,"change".getBytes());
        } catch (Exception e) {
//            e.printStackTrace();
            ZKExceptionHandler.handlerZKException(e);
        }
    }

    //处理状态枚举
    private TaskStateEnum judgeState(short state) {
        TaskStateEnum taskState = null;
        switch (state) {
            case TaskService.STATE_READY: {
                taskState = TaskStateEnum.STATE_READY;
                break;
            }
            case TaskService.STATE_RUNNING: {
                taskState = TaskStateEnum.STATE_RUNNING;
                break;
            }
            case TaskService.STATE_SUCCESS: {
                taskState = TaskStateEnum.STATE_SUCCESS;
                break;
            }
            case TaskService.STATE_WORKER_INNER_FAIL: {
                taskState = TaskStateEnum.STATE_WORKER_INNER_FAIL;
                break;
            }
            case TaskService.STATE_WORKER_OUTER_FAIL: {
                taskState = TaskStateEnum.STATE_WORKER_OUTER_FAIL;
                break;
            }
            default:
                break;
        }
        return taskState;
    }

    //封装任务与任务日志类
    private void resolveTaskJson(Task task, TaskLog taskLog, TaskStateEnum state,short[] states) throws Exception {
        Map<String, Object> taskMap = new HashMap<>();
        taskMap.put("stateDesc",state.stateDesc());
        taskMap.put("state",state.getState());
        taskMap.put("id", task.getId());
        taskMap.put("states", states);

        int count = taskService.updateTaskByStates(taskMap);
        if(count > 0) {
            taskLog.setStatus(state.getState());
            taskLogService.addTaskLog(taskLog);
        }
    }
}
